package com.zyx.kafkademo.producer;


import com.zyx.kafkademo.utils.KafkaUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author zyx
 * @since 2021/04/20 0:30
 */
public class ProducerDemo {
    public static void main(String[] args) throws InterruptedException {
        KafkaProducer<String, String> kafkaProducer = KafkaUtils.getKafkaProdocuer();

        Random random = new Random();
        String[] words = {"你好", "World", "中文", "Flink", "世界"};
        // String[] words = {"hello", "World", "chinese", "Flink", "spark"};
        int wordsLen = words.length;

        for (int k = 0; k < 10; k++) {
            for (int i = 0; i < 100; i++) {

                StringBuilder sb = new StringBuilder();
                for (int j = 0; j <= 2 + random.nextInt(3); j++) {
                    sb.append(words[random.nextInt(wordsLen)]).append(" ");
                }
                String line = sb.subSequence(0, sb.length() - 1).toString();
                ProducerRecord<String, String> record = new ProducerRecord<>("kafka_topic_wc", line);
                kafkaProducer.send(record);
                System.out.println("producer >>" + line);
            }
            TimeUnit.SECONDS.sleep(2L);
        }

        kafkaProducer.close();
    }
}
